//package read.niit.service
//
//
//import org.apache.spark.rdd.RDD
//import org.apache.spark.streaming.dstream.DStream
//import read.niit.bean.Reader
//
//class ReaderData {
//
//  def dataAnalysis(reader: DStream[Reader]): Unit = {
//
//    // hotRead_subjectTop10(reader)
//    //    studentWithMaxScoreBookTop10(reader)
//    realTimeByOwnDefinedTime(reader)
//
//  }
//
//
//
//
//  //需求六：统计每个学生的得分最低的前10道题，并带上所属题
//  private def studentWithMaxScoreBookTop10(reader: DStream[Reader]): Unit = {
//    //(学生,(题目1,成绩),(题目2,成绩),(题目3,成绩),…………(题目10,成绩))
//    val mapDS = reader.map(data => {
//      (data.student_id, (data.book_id, data.read_subject_id, data.score))
//    })
//
//    val groupDS = mapDS.groupByKey()
//
//    groupDS.foreachRDD(rdd => {
//      val resRDD = rdd.mapValues(iter => {
//        val sort = iter.toList.sortBy(_._2)
//        sort.take(10)
//      })
//      println("--------统计每本书籍评分最高的前8本，并带上所属书----------")
//      resRDD.collect().foreach(println)
//    })
//
//  }
//
//  private def realTimeByOwnDefinedTime(reader: DStream[Reader]): Unit = {
//    val mapDS: DStream[Any] = reader.map(data => {
//      val timeOfOwnDefined = data.date_id.split("_")(1)
//      if (timeOfOwnDefined.equals("4.23")) {
//        (data.date_id, (data.grade_id, data.read_subject_id))
//      } else {
//        null
//      }
//    })
//    mapDS.foreachRDD(rdd => {
//      val result: RDD[Any] = rdd.filter(rdd => rdd != null)
//      println("===========4.23===========")
//      result.collect().foreach(println)
//
//
//    })
//  }
//}
//
////需求一：统计top10阅读书籍类型
////    private def hotRead_subjectTop10(answer: DStream[Reader]):Unit ={
////      // 将结果在控制台打印
////      // k-v (书籍，1)
////      val mapDS = answer.map(data => {
////        (data.read_subject_id,1)
////      })
////      //(书籍，1) (书籍，1) (书籍，1) ==> (书籍，3)
////      val reduceData = mapDS.reduceByKey(_ + _)
////      // sparkStreaming 先sort再输出  先输出再排序
////      // 输出 print  foreachRDD
////      reduceData.foreachRDD((rdd=>{
////        // 第一个下划线：(书籍，3)  _2:次数  元组索引值从1开始  默认是降序
////        val sortRDD:RDD[(String,Int)] = rdd.sortBy(_._2,false)
////        val top10:Array[(String,Int)] = sortRDD.take(10)
////        top10.foreach(println)
////        println("---------------------------------------")
////      }))
////    }
